package com.whz.bus.debezium;

import io.debezium.embedded.Connect;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.RecordChangeEvent;
import io.debezium.engine.format.ChangeEventFormat;
import io.debezium.engine.format.Json;
import lombok.Builder;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.connect.source.SourceRecord;
import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;
import tech.tongyu.common.utils.spring.SpringUtils;

/**
 * debezium引擎创建工厂
 *
 * @author whz
 */
@Slf4j
@Builder
public class DebeziumEngineFactory {
    /**
     * 数据库配置
     */
    private DataSourceProperties dataSourceProperties;
    /**
     * debezium 配置
     */
    private DebeziumProperties debeziumProperties;

    /**
     * 链接关闭后回调
     */
    @Builder.Default
    private DebeziumEngine.CompletionCallback completionCallback = new TmcCompletionCallback();

    @Builder.Default
    private DebeziumEngine.ConnectorCallback connectorCallback = new TmcConnectorCallback();


    /**
     * 创建Kafka api风格的事件数据类型
     *
     * @param changeRecordConsumer Kafka api风格数据类型的消费者
     * @return debezium 引擎
     */
    public DebeziumEngine<RecordChangeEvent<SourceRecord>> createKafkaConnectApiEngine(DebeziumEngine.ChangeConsumer<RecordChangeEvent<SourceRecord>> changeRecordConsumer) {
        final DebeziumConfigurationWrapper debeziumConfigurationWrapper = getDebeziumConfigurationWrapper();
        return DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
                .using(debeziumConfigurationWrapper.debeziumConfig().asProperties())
                .using(ClassUtils.getDefaultClassLoader())
                .using(completionCallback)
                .using(connectorCallback)
                .notifying(changeRecordConsumer)
                .build();
    }

    /**
     * 创建json风格事件数据类型
     *
     * @param consumer json风格的数据消费者
     * @return debezium 引擎
     */
    public DebeziumEngine<RecordChangeEvent<String>> createJsonApiEngine(DebeziumEngine.ChangeConsumer<RecordChangeEvent<String>> consumer) {
        final DebeziumConfigurationWrapper debeziumConfigurationWrapper = getDebeziumConfigurationWrapper();
        return DebeziumEngine.create(ChangeEventFormat.of(Json.class))
                .using(debeziumConfigurationWrapper.debeziumConfig().asProperties())
                .using(ClassUtils.getDefaultClassLoader())
                .using(completionCallback)
                .using(connectorCallback)
                .notifying(consumer)
                .build();
    }

    private DebeziumConfigurationWrapper getDebeziumConfigurationWrapper() {
        Assert.notNull(debeziumProperties, "debezium factory need debeziumProperties");
        return new DebeziumConfigurationWrapper(debeziumProperties, getDataSourceWrapper());
    }

    private DataSourceWrapper getDataSourceWrapper() {
        Assert.notNull(dataSourceProperties, "debezium factory need dataSourceProperties.");
        return new DataSourceWrapper(dataSourceProperties);
    }

    private static class TmcConnectorCallback implements DebeziumEngine.ConnectorCallback {
        @Override
        public void connectorStarted() {
            log.debug("connectorStarted");
        }

        @Override
        public void connectorStopped() {
            log.debug("connectorStopped");
        }

        @Override
        public void taskStarted() {
            log.debug("taskStarted");
        }

        @Override
        public void taskStopped() {
            log.debug("taskStopped");
        }
    }

    private static class TmcCompletionCallback implements DebeziumEngine.CompletionCallback {

        @Override
        public void handle(final boolean success, final String message, final Throwable error) {
            if (success) {
                log.info(message);
            } else {
                log.error(message, error);
            }
            SpringUtils.getApplicationContext().ifPresent(applicationContext -> applicationContext.publishEvent(new DebeziumEvent(this, message)));
        }
    }
}